package com.hugedata.wpm.service.action;

import java.io.ByteArrayInputStream;
import java.io.File;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.ResourceBundle;
import java.util.concurrent.ConcurrentHashMap;
import java.util.regex.Pattern;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.time.DateFormatUtils;
import org.apache.commons.lang.time.DateUtils;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.http.HttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.InputStreamEntity;
import org.apache.http.impl.client.DefaultHttpClient;
import org.apache.log4j.Logger;

import com.google.gson.Gson;
import com.hugedata.oss.logger.OSSLogPriority;
import com.hugedata.wpm.service.TaskService;
import com.hugedata.wpm.service.bean.MonitorCfg;
import com.hugedata.wpm.service.bean.NodeBean;
import com.hugedata.wpm.service.bean.report.ResultTask;
import com.hugedata.wpm.service.bean.report.TaskNodes;
import com.hugedata.wpm.startup.WPMRealtime;
import com.hugedata.wpm.util.Constants;
import com.hugedata.wpm.util.Sender;
import com.hugedata.wpm.util.SystemUtils;
import com.hugedata.wpm.util.WpmException;

public class DataCompleteJob
{
    private static Logger logger = Logger.getLogger(DataCompleteJob.class);
    private static Logger ossLogger = Logger.getLogger(DataCompleteJob.class);
    private static ResourceBundle rb = ResourceBundle.getBundle("oss_alarm");// 绑定资源文件

//    private DateFormat dfd = new SimpleDateFormat("yyyyMMdd");
//    private DateFormat dft = new SimpleDateFormat("HHmm");
//    private DateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
//    private DateFormat adddf = new SimpleDateFormat("yyyyMMddHHmmss");

    // 当补数据次数达到3次则告警
    // public final static int countnum = 3;
    // 节点补数据存放全局map
    public static Map<String, String> taskMailContent = new HashMap<String, String>();
    public static Map<String, ConcurrentHashMap<String, Integer>> nummap = new ConcurrentHashMap<String, ConcurrentHashMap<String, Integer>>();
    // 连续告警时间记录表
    public static Map<String, Long> continuousTimeMap = new ConcurrentHashMap<String, Long>();
    // 控制对task缺失上报数据时的发送邮件时间
    public static Map<String, Long> mailTimeMap = new ConcurrentHashMap<String, Long>();

    private TaskService taskService;
    private Sender sender;
    private int taskadddateTimes;
    private Gson gson;
    private String dataPostUrl;
    private int isUseFile;
    private int sendCycle;

    public int getSendCycle()
    {
        return sendCycle;
    }

    public void setSendCycle(int sendCycle)
    {
        this.sendCycle = sendCycle;
    }

    public String getDataPostUrl()
    {
        return dataPostUrl;
    }

    public void setDataPostUrl(String dataPostUrl)
    {
        this.dataPostUrl = dataPostUrl;
    }

    public int getIsUseFile()
    {
        return isUseFile;
    }

    public void setIsUseFile(int isUseFile)
    {
        this.isUseFile = isUseFile;
    }

    public int getTaskadddateTimes()
    {
        return taskadddateTimes;
    }

    public void setTaskadddateTimes(int taskadddateTimes)
    {
        this.taskadddateTimes = taskadddateTimes;
    }

    public Gson getGson()
    {
        return gson;
    }

    public void setGson(Gson gson)
    {
        this.gson = gson;
    }

    public String getGetDataUrl()
    {
        return getDataUrl;
    }

    public void setGetDataUrl(String getDataUrl)
    {
        this.getDataUrl = getDataUrl;
    }

    public int getDelayCycle()
    {
        return delayCycle;
    }

    public void setDelayCycle(int delayCycle)
    {
        this.delayCycle = delayCycle;
    }

    // 配置获取数据的路径
    private String getDataUrl;

    // 配置延时周期 补数据延迟周期
    private int delayCycle;

    public Sender getSender()
    {
        return sender;
    }

    public void setSender(Sender sender)
    {
        this.sender = sender;
    }

    public TaskService getTaskService()
    {
        return taskService;
    }

    public void setTaskService(TaskService taskService)
    {
        this.taskService = taskService;
    }

    /*
     * 判断是否为整数
     * 
     * @param str 传入的字符串
     * 
     * @return 是整数返回true,否则返回false
     */

    public static boolean isInteger(String str)
    {
        Pattern pattern = Pattern.compile("^[-\\+]?[\\d]*\\.0$");
        return pattern.matcher(str).matches();
    }

    // 查询告警人员信息，根据配置下发相应的信息
    public void sendSmsMail(String context, String context2)
    {

        List<MonitorCfg> mCfgs;
        try
        {
            mCfgs = taskService.findMoniterUsers();
            for (MonitorCfg m : mCfgs)
            {
                String[] alarmobjects =
                { m.getAlarmObj() };
                if (Constants.NUM_1.equals(m.getAlarmObjType()))
                {
                    if (!"".equals(context2))
                    {
                        int ret = sender.sendSmsWithSoap(0, 0, alarmobjects,
                                context2, 2, 1, "0");
                        if (ret == 0)
                        {
                            logger.info("datacomplete send sms to "
                                    + alarmobjects[0] + "success,content:"
                                    + context2);
                        }
                        else
                        {
                            logger.info("datacomplete send sms to "
                                    + alarmobjects[0] + "failed,content:"
                                    + context2);
                        }
                    }
                }
                else if (Constants.NUM_2.equals(m.getAlarmObjType()))
                {
                    if (!"".equals(context))
                    {
                        StringBuffer content = new StringBuffer();
                        content.append("<p>" + context + "</p>");
                        String subject = StringUtils.substring(context, 0, 50)
                                + "......";
                        int ret = sender.sendMailWithSoap("0", alarmobjects,
                                subject, content.toString());
                        // 返回-1为异常
                        if (ret == 0)
                        {
                            logger.info("datacomplete send sms to "
                                    + alarmobjects[0] + "success,content:"
                                    + context);
                        }
                        else
                        {
                            logger.info("datacomplete send sms to "
                                    + alarmobjects[0] + "failed,content:"
                                    + context);
                        }
                    }
                }
                else
                {
                    logger.error("illegel alarm_obj_type[1=sms,2=mail] is "
                            + m.getAlarmObjType());
                }
            }
        }
        catch (WpmException e)
        {
            ossLogger.log(OSSLogPriority.OSS_ERROR_LEVEL, String.format(
                    rb.getString("10120001"), "DataCompleteJob发邮件或者短信"));
            logger.error("DataCompleteJob SmsMailWorker error:" + e.toString());
        }

    }

    public String getSmsContent(String taskId, String imspTaskId, String content)
    {
        StringBuilder smsSb = new StringBuilder();
        smsSb.append("网页监测项目：业务ID：").append(taskId).append(",测量节点工具任务ID：")
                .append(imspTaskId).append(",");
        smsSb.append(content);
        return smsSb.toString();
    }

    public String getMailContent(String taskId, String imspTaskId,
            List<String> nodeIdList, String content)
    {
        StringBuilder mailSb = new StringBuilder();
        mailSb.append("网页监测业务ID：").append(taskId).append("，工具ID：")
                .append(imspTaskId).append(",下发失败节点为：");
        for (String nodeId : nodeIdList)
        {
            mailSb.append(nodeId + ",");
        }
        mailSb.append("其中").append(content);
        return mailSb.toString();
    }

    public String getLogContent(String taskId, String imspTaskId,
            List<String> nodeIdList, String content)
    {
        StringBuilder sb = new StringBuilder();
        sb.append("WPM_Task:").append(taskId).append(",imspTaskId:")
                .append(imspTaskId).append(",NodeIds:");
        for (String nodeId : nodeIdList)
        {
            sb.append("[" + nodeId + "]");
        }
        sb.append(",").append(content);
        return sb.toString();
    }

    public int getRandomIndex(int length)
    {
        return (int) (Math.random() * length);
    }

    public static final String YYYYMMDD = "yyyyMMdd";
    public static final String HHMM = "HHmm";
//    public static final String YYYY_MM_DD_HH_MM_SS = "yyyy-MM-dd HH:mm:ss";
    public static final String YYYYMMDDHHMMSS = "yyyyMMddHHmmss";
    
    public void execute()
    {
        logger.info("Data Complete Scheduled Scans Start.....");
        HTable table = null;
        try
        {
            // 查询所有符号条件的任务(网页监测已经支付的任务)
            // 并查询该任务所有nodeids
            List<TaskNodes> taskNodes = (List<TaskNodes>) taskService
                    .findTaskNodes();
            logger.info("TaskNodes meet the requirement in mysql is "
                    + taskNodes.size());
            
            if (null == taskNodes 
                    || taskNodes.size() == 0)
            {
                logger.info("there is no taskNodes");
                return;
            }

            // 一次扫描周期内的全部resultTask
            ArrayList<ResultTask> datas = new ArrayList<ResultTask>();

            Date currentDate = new Date();
            String taskDay = DateFormatUtils.format(currentDate, YYYYMMDD);
            String taskHourMinute = DateFormatUtils.format(currentDate, HHMM);
            logger.info("taskDay:" + taskDay + " " + "taskHourMinute:"
                    + taskHourMinute);
            
            String resetTime = "";// 补数据的时间
            Calendar calendar = Calendar.getInstance();

            table = new HTable(Constants.HBASE_CONF,Constants.TABLE_REALTIME);

            for (TaskNodes aTaskNodes : taskNodes)
            {
                String taskId = aTaskNodes.getTaskID();
                int taskInterval = aTaskNodes.getTaskInterval();
                String imspTaskId = aTaskNodes.getImspTaskId();
                imspTaskId = StringUtils.trimToEmpty(imspTaskId);
                
                // 获得一个task的全部nodeId
                List<String> nodeIdList = new ArrayList<String>();
                List<String> nodes = aTaskNodes.getNodes();
                for (String nodeString : nodes)
                {
                    if (StringUtils.isNotBlank(nodeString))
                    {
                        String[] nodeArray = StringUtils.split(nodeString, "_");
                        for (String node : nodeArray)
                        {
                            if (StringUtils.isNotBlank(node))
                            {
                                nodeIdList.add(node);
                            }
                        }
                    }
                }
                
                // 任务的开始时间用来后期补数据确定时间
                // 判断resume字段是否为空，空取starttime作为补数据开始时间，否则用resume
                String beginTime;
                String resumeTmp = aTaskNodes.getResume();
                if (StringUtils.isNotBlank(resumeTmp))
                {
                    beginTime = aTaskNodes.getResume();
                }
                else
                {
                    beginTime = aTaskNodes.getStarttime();
                }

                Date taskBeginDate = DateUtils.parseDate(beginTime, new String[]{"yyyy-MM-dd HH:mm:ss"});

                long ll = (currentDate.getTime() - taskBeginDate.getTime())
                        / (60 * 1000);// 计算出当前时间距离开始时间相差多少分钟
                
                double timenu = (double) ll / (double) taskInterval;

                if (ll >= delayCycle * taskInterval)
                {
                    // 判断文件路径取前delayCycle个周期前的日期
//                    long taskdaytime = currentDate.getTime() - delayCycle
//                            * taskInterval * 60 * 1000;
//                    Date ttime = new Date(taskdaytime);
//                    String taskDayt = dfd.format(ttime);
                    
                    Date beforeDate = DateUtils.addMinutes(currentDate, -(delayCycle * taskInterval));
                    String beforeDateDayStr = DateFormatUtils.format(beforeDate, YYYYMMDD);
                    
                    String filePathParent = getDataUrl + File.separator
                            + beforeDateDayStr + File.separator + taskId
                            + File.separator;
                    File taskDir = new File(filePathParent);
                    if (!taskDir.exists())
                    {
                        String taskBeginDateDayStr = DateFormatUtils.format(taskBeginDate, YYYYMMDD);
                        // 新创建的任务已过第一个周期且没有上报数据
                        if (((int) timenu) > delayCycle
                                && taskBeginDateDayStr.equals(taskDay))
                        {
                            String enText = "is built today,but still miss report message after its first cycle";
                            String cnText = "已过第一个周期且没有上报数据,请核查！";
                            logger.info(getLogContent(taskId, imspTaskId,
                                    nodeIdList, enText));
                            if (mailTimeMap.containsKey(taskId))
                            {
                                if ((currentDate.getTime() - mailTimeMap
                                        .get(taskId)) > 60 * 60 * 1000)
                                {
                                    sendSmsMail(
                                            getMailContent(taskId,
                                                    imspTaskId, nodeIdList,
                                                    cnText),
                                            getSmsContent(taskId,
                                                    imspTaskId, cnText));
                                    mailTimeMap.put(taskId, currentDate.getTime());
                                }
                            }
                            else
                            {
                                mailTimeMap.put(taskId, currentDate.getTime());
                                sendSmsMail(
                                        getMailContent(taskId, imspTaskId,
                                                nodeIdList, cnText),
                                        getSmsContent(taskId, imspTaskId,
                                                cnText));
                            }
                            continue;
                        }
                        
//                        Date yesterdayDate = new Date();
//                        yesterdayDate.setTime(yesterdayDate.getTime() - 24
//                                * 60 * 60 * 1000);
//                        String taskDateYesterday = dfd
//                                .format(yesterdayDate);
                        
                        Date yesterdayDate = DateUtils.addDays(Calendar.getInstance().getTime(), -1);
                        String yesterDayDateDayStr = DateFormatUtils.format(yesterdayDate, YYYYMMDD);
                        
                        filePathParent = getDataUrl + File.separator
                                + yesterDayDateDayStr + File.separator
                                + taskId + File.separator;
                        
                        File yesterdayTaskDir = new File(filePathParent);
                        if (!yesterdayTaskDir.exists())
                        {
//                            yesterdayDate.setTime(yesterdayDate.getTime()
//                                    - 24 * 60 * 60 * 1000);
//                            taskDateYesterday = dfd.format(yesterdayDate);
                            Date beforeYesterdayDate = DateUtils.addDays(yesterdayDate, -1);
                            String beforeYesterdayDateDayStr = DateFormatUtils.format(beforeYesterdayDate, YYYYMMDD);
                            
                            filePathParent = getDataUrl + File.separator
                                    + beforeYesterdayDateDayStr + File.separator
                                    + taskId + File.separator;
                            yesterdayTaskDir = new File(filePathParent);
                            if (!yesterdayTaskDir.exists())
                            {
                                String enText = "missed all cache data for consecutive days";
                                String cnText = "连续多天没有上报数据,请核查！";
                                logger.info(getLogContent(taskId,
                                        imspTaskId, nodeIdList, enText));

                                if (mailTimeMap.containsKey(taskId))
                                {
                                    if ((currentDate.getTime() - mailTimeMap
                                            .get(taskId)) > 60 * 60 * 1000)
                                    {
                                        sendSmsMail(
                                                getMailContent(taskId,
                                                        imspTaskId,
                                                        nodeIdList, cnText),
                                                getSmsContent(taskId,
                                                        imspTaskId, cnText));
                                        mailTimeMap.put(taskId,
                                                currentDate.getTime());
                                    }
                                }
                                else
                                {
                                    mailTimeMap.put(taskId, currentDate.getTime());
                                    sendSmsMail(
                                            getMailContent(taskId,
                                                    imspTaskId, nodeIdList,
                                                    cnText),
                                            getSmsContent(taskId,
                                                    imspTaskId, cnText));
                                }
                                continue;
                            }
                            String enText = "missed all cache data in last two days,"
                                    + "data complete will use message from the day before yesterday";
                            logger.info(getLogContent(taskId, imspTaskId,
                                    nodeIdList, enText));
                        }
                        String enText = "missed all cache data today,"
                                + "data complete will use message from yesterday";
                        logger.info(getLogContent(taskId, imspTaskId,
                                nodeIdList, enText));
                    }
                    
                    
                    String startTime = "";
                    String endTime = "";
                    // 整数时刚好在周期上，需要取左右的时间
                    if (isInteger(String.valueOf(timenu)))
                    {
                        // 重新设置开始时间结束时间
                        calendar.setTime(currentDate);
                        calendar.add(Calendar.MINUTE, 0 - (delayCycle + 1)
                                * taskInterval + 1);
                        startTime = dft.format(calendar.getTime());

                        calendar.setTime(currentDate);
                        calendar.add(Calendar.MINUTE, 0 - delayCycle
                                * taskInterval + 1);
                        endTime = dft.format(calendar.getTime());
                    }
                    else
                    // 不在整数时刚好在2个周期的区间上
                    {
                        calendar.setTime(currentDate);
                        calendar.add(Calendar.MINUTE, 0 - (delayCycle + 1)
                                * taskInterval);
                        startTime = dft.format(calendar.getTime());

                        calendar.setTime(currentDate);
                        calendar.add(Calendar.MINUTE, 0 - delayCycle
                                * taskInterval);
                        endTime = dft.format(calendar.getTime());
                    }

                    // 查询条件endRow的时间taskTime是当前时间
                    // startRow的时间startTime则为当前时间-任务间隔时间
//                    String endRow = taskDayt + "_" + taskId + "_"
//                            + taskHourMinute;
//                    String startRow = taskDayt + "_" + taskId + "_"
//                            + startTime;
                    
                    //beforeDateDayStr的格式为：YYYYMMDD
                    String startRow = beforeDateDayStr + "_" + taskId + "_"
                            + startTime;
                    String endRow = beforeDateDayStr + "_" + taskId + "_"
                            + endTime;
                    
                    logger.info("HBase Scan startRow:" + startRow);
                    logger.info("HBase Scan endRow:" + endRow);
                    
                    // 根据yyyyMMdd_taskid_time拼装scan查询WPM_Realtime表
                    Scan scan = new Scan();
                    scan.setStartRow(Bytes.toBytes(startRow));
                    scan.setStopRow(Bytes.toBytes(endRow));
                    scan.setBatch(100);
                    scan.setCaching(1);
                    ResultScanner scanner = table.getScanner(scan);

                    // 查询WPM_Realtime表中真正已经上报数据的nodeid
                    // 并从nodeIdList中移除上报过的nodeId
                    // 则nodeIdList中最后剩下的nodeId为缺失的nodeId
                    for (Result r : scanner)
                    {
                        String rowKey = Bytes.toString(r.getRow());
                       
                        String nodeId = "";
                        nodeId = StringUtils.substring(rowKey,
                                rowKey.length() - 4);

                        if (nodeIdList.contains(nodeId))
                        {
                            nodeIdList.remove(nodeId);
                        }
                    }

                    while (nodeIdList.contains(null))
                    {
                        nodeIdList.remove(null);
                    }
                    logger.info("NodeIdList length is " + nodeIdList.size());

                    if (0 == nodeIdList.size())
                    {
                        nummap.remove(aTaskNodes.getTaskID());
                    }

                    StringBuilder cnText = new StringBuilder();

                    // 根据缺失的nodeId从对应的本地目录中获得该nodeId的最新报文
                    for (String nodeId : nodeIdList)
                    {
                        String filePath = filePathParent + nodeId;

                        File fileDir = new File(filePath);
                        File lastFile = null;

                        // 若该nodeId在本地没有缓存
                        // 则获取该任务其它nodeId下的最新报文
                        if (!fileDir.exists())
                        {
                            File[] nodeFiles = fileDir.getParentFile()
                                    .listFiles();
                            if (null == nodeFiles)
                            {
                                continue;
                            }
                            else
                            {
                                int index = getRandomIndex(nodeFiles.length);
                                filePath = filePathParent
                                        + nodeFiles[index].getName();
                                nodeId = nodeFiles[index].getName();
                            }
                        }

                        // 确定需要补数据的时间
                        // 当前时间-开始时间 /频率
                        // 算出在那个周期之间，然后从开始时间+频率*周期得到上报的时间点
                        logger.info("WPM_Task:"
                                + taskId
                                + " nodeid:"
                                + nodeId
                                + " is doing data completion,data is from dir:"
                                + filePath);

                        // 从开始时间向前推num个周期得到当前补数据的时间
                        long numlong = (long) timenu - delayCycle;
                        long numtime = taskBeginDate.getTime() + numlong
                                * taskInterval * 60 * 1000;
                        Date dates = new Date(numtime);
                        resetTime = adddf.format(dates);

                        fileDir = new File(filePath);
                        File[] files = fileDir.listFiles();
                        lastFile = files[files.length - 1];
                        // 读取文件内容
                        String src = com.hugedata.wpm.util.SystemUtils
                                .readDateFromPath(lastFile);

                        // 替换最新报文中的loadStartTime与moniteringTime以及nodeid
                        src = src.replaceAll(
                                "\"monitoringTime\":\"(\\d{14})\"",
                                "\"monitoringTime\":\"" + resetTime + "\"");
                        src = src.replaceAll(
                                "\"loadStartTime\":\"(\\d{14})\"",
                                "\"loadStartTime\":\"" + resetTime + "\"");
                        src = src.replaceAll("\"nodeID\":\"(\\d{4})\"",
                                "\"nodeID\":\"" + nodeId + "\"");
                        // 计算器存放map中，当大于设置的数字报警
                        // 同一个任务链接补数据

                        if (nummap.containsKey(taskId))
                        {
                            ConcurrentHashMap<String, Integer> map = nummap
                                    .get(taskId);

                            // 判断本次的节点上次是否出现，木有就删除，保证连续几次出现的准确性
                            for (Map.Entry<String, Integer> entry : map
                                    .entrySet())
                            {

                                if (!nodeIdList.contains(entry.getKey()))
                                {
                                    map.remove(entry.getKey());
                                }
                            }

                            if (map.containsKey(nodeId))// 上次保存的节点信息
                            {
                                int num = map.get(nodeId);
                                if (num + 1 == taskadddateTimes)
                                {
                                    // 报警
                                    NodeBean bean = (NodeBean) taskService
                                            .findNodeinfoByNodeID(nodeId);
                                    if (null != bean
                                            && null != bean
                                                    .getDescription()
                                            && !"".equals(bean
                                                    .getDescription()))
                                    {
                                        cnText.append(bean.getDescription())
                                                .append("(" + nodeId + ")")
                                                .append(",");
                                        String enText = "node:"
                                                + nodeId
                                                + ",Continuous occurrences data comletion";
                                        logger.info(getLogContent(taskId,
                                                imspTaskId, nodeIdList,
                                                enText));
                                    }
                                    // 报警完删除 重新开始
                                    map.remove(nodeId);
                                }
                                else
                                {
                                    map.put(nodeId, num + 1);
                                }
                            }
                            else
                            // 新增的节点信息
                            {
                                map.put(nodeId, 1);
                                nummap.put(taskId, map);
                            }

                        }
                        else
                        {
                            ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<String, Integer>();
                            map.put(nodeId, 1);
                            nummap.put(taskId, map);
                        }
                        // 获得替换后的报文，gson解析为对象并加入列表中
                        ResultTask resultTask = (ResultTask) gson.fromJson(
                                src, ResultTask.class);
                        if (1 == isUseFile)
                        {
                            StringBuffer sb = new StringBuffer(
                                    "{\"clientID\":");
                            sb.append(nodeId).append(",\"timestamp\":\"")
                                    .append(dates.getTime())
                                    .append("\",\"data\":[").append(src)
                                    .append("]}");
                            HttpPost post = new HttpPost(dataPostUrl);
                            //新修改上报为压缩文件
                            post.setHeader("Content-Type", "application/json; charset=UTF-8");
                            post.setHeader("Accept-Encoding", "gzip");
                            byte[] bgzip = SystemUtils.gzip(sb.toString());
                            
                            DefaultHttpClient client = new DefaultHttpClient();
//                            HttpEntity re = new StringEntity(sb.toString(),
//                                    HTTP.UTF_8);
                            InputStreamEntity httpentity = new InputStreamEntity(
                                    new ByteArrayInputStream(bgzip), bgzip.length);
                            httpentity.setChunked(true);
                            post.setEntity(httpentity);

//                            post.getParams().setParameter("Content-Type",
//                                    "text/html;charset=UTF-8");
                            HttpResponse response = client.execute(post);
                            logger.info("Post response:"
                                    + response.getStatusLine()
                                            .getStatusCode());
                        }
                        datas.add(resultTask);
                        logger.info("WPM_Task:" + taskId + " nodeId:"
                                + nodeId
                                + " data completion change time to:"
                                + resetTime);

                    }
                    if (cnText.length() > 0)
                    {
                        cnText.append("节点连续报警，请核查！<br/>");
                        String content = "";
                        if (taskMailContent.containsKey(taskId))
                        {
                            content = taskMailContent.get(taskId);
                        }
                        content += df.format(currentDate)
                                + ","
                                + getMailContent(taskId, imspTaskId,
                                        nodeIdList, cnText.toString());
                        taskMailContent.put(taskId, content);
                    }
                    if (continuousTimeMap.containsKey(taskId))
                    {
                        if ((currentDate.getTime() - continuousTimeMap.get(taskId)) > sendCycle * 60 * 1000)
                        {
                            String content = "";
                            if (taskMailContent.containsKey(taskId))
                            {
                                content = taskMailContent.get(taskId);
                            }
                            sendSmsMail(content, "");
                            continuousTimeMap.put(taskId, currentDate.getTime());
                            taskMailContent.remove(taskId);
                        }
                    }
                    else
                    {
                        continuousTimeMap.put(taskId, currentDate.getTime());
                    }
                }
                else
                {
                    logger.info("WPM_Task:"
                            + taskId
                            + ",current cycle out of delay cycle,do not excute!");
                }
            }
            // 一个扫描周期结束后把ResultTask队列中的ResultTask加入到
            // Hbase表中，完成数据补全
            if (0 == isUseFile && datas.size() > 0)
            {
                WPMRealtime.insertRealtimes(datas, 0L);
                WPMRealtime.insertRealtimeDetails(datas, 0L);
            }
            logger.info("Final ResultTasks'size Insert into HBase:"
                    + datas.size());
        
        }
        finally
        {
            table.close();
        }
    }
}
